package app.dwd;

import app.ods.Constant;
import app.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * DWD层：用户行为数据清洗与权重扩充
 * 核心逻辑：关联商品属性表，计算商品类别权重、行为权重、时间衰减系数
 */
public class UserBehaviorEnrichment {
    public static void main(String[] args) throws Exception {
        // 1. 初始化Flink执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); // 测试阶段设置并行度为1，生产环境按集群规模调整
        EnvironmentSettings tableEnvSettings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, tableEnvSettings);

        // 2. 注册Kafka ODS层数据源表（用户行为原始数据）
        tableEnv.executeSql("CREATE TABLE ods_user_behavior (" +
                "  user_id STRING COMMENT '用户ID'," +
                "  goods_id STRING COMMENT '商品ID'," +
                "  behavior_type STRING COMMENT '行为类型：purchase/search/collect/browse'," +
                "  behavior_time BIGINT COMMENT '行为时间（毫秒）'," +
                "  search_keyword STRING COMMENT '搜索关键词（仅搜索行为有值）'," +
                "  goods_category STRING COMMENT '商品类目（如：童装/婴儿装/亲子装>女童连衣裙）'," +
                "  gender_weight DOUBLE COMMENT '商品性别权重（1.0=女宝/男宝指向，0.5=中性，0.2=未指向）'," +
                "  gender_direction STRING COMMENT '性别指向（female/male/neutral/none）'," +
                "  age_stage STRING COMMENT '关联年龄分段（如：0-3个月）'," +
                "  age_weight DOUBLE COMMENT '年龄分段权重（1.0=强关联，0.8=中关联，0.5=弱关联）'," +
                "  ts AS TO_TIMESTAMP_LTZ(behavior_time, 3) COMMENT '行为时间戳（Flink时间属性）'," +
                "  WATERMARK FOR ts AS ts - INTERVAL '5' SECOND COMMENT '水位线：允许5秒延迟'" +
                ") "+ MyKafkaUtil.getKafkaDDL("user_biao_topic", "user_biao_topic_1565"));

        // 4. 注册Kafka DWD层输出表（扩充权重后的数据）
        tableEnv.executeSql("CREATE TABLE dwd_user_behavior_enriched (" +
                "  user_id STRING COMMENT '用户ID'," +
                "  goods_id STRING COMMENT '商品ID'," +//
                "  behavior_type STRING COMMENT '行为类型'," +
                "  behavior_time BIGINT COMMENT '行为时间（毫秒）'," +
                "  search_keyword STRING COMMENT '搜索关键词'," +
                "  goods_category STRING COMMENT '商品类目'," +
                "  gender_weight DOUBLE COMMENT '商品性别权重'," +
                "  gender_direction STRING COMMENT '商品性别指向'," +
                "  age_stage STRING COMMENT '商品关联年龄分段'," +
                "  age_weight DOUBLE COMMENT '商品年龄权重'," +
                "  behavior_weight DOUBLE COMMENT '行为权重'," +
                "  time_decay DOUBLE COMMENT '时间衰减系数'," +
                "  ts TIMESTAMP_LTZ(3) COMMENT '行为时间戳'," +
                "  WATERMARK FOR ts AS ts - INTERVAL '5' SECOND" +
                ")"+ MyKafkaUtil
                .getKafkaDDL("dwd_user_behavior_enriched", "dwd_user_behavior_enriched_1565"));

        // 5. 核心SQL：数据清洗与权重计算（DWD层核心逻辑）
        String dwdEnrichmentSql = "INSERT INTO dwd_user_behavior_enriched " +
                "SELECT " +
                "  a.user_id," +
                "  a.goods_id," +//
                "  a.behavior_type," +
                "  a.behavior_time," +
                "  a.search_keyword," +
                "  a.goods_category," +
                "  -- 商品性别权重：关联商品属性表，无匹配时默认0.2（未指向）" +
                "  COALESCE(b.gender_weight, 0.2) AS gender_weight," +
                "  COALESCE(b.gender_direction, 'none') AS gender_direction," +
                "  COALESCE(b.age_stage, '未知') AS age_stage," +
                "  COALESCE(b.age_weight, 0.5) AS age_weight," +
                "  -- 行为权重：根据行为类型赋值" +
                "  CASE " +
                "    WHEN a.behavior_type = '" + Constant.BEHAVIOR_TYPE_PURCHASE + "' THEN " + Constant.WEIGHT_PURCHASE +
                "    WHEN a.behavior_type = '" + Constant.BEHAVIOR_TYPE_SEARCH + "' THEN " + Constant.WEIGHT_SEARCH +
                "    WHEN a.behavior_type = '" + Constant.BEHAVIOR_TYPE_COLLECT + "' THEN " + Constant.WEIGHT_COLLECT +
                "    WHEN a.behavior_type = '" + Constant.BEHAVIOR_TYPE_BROWSE + "' THEN " + Constant.WEIGHT_BROWSE +
                "    ELSE 0.0 " +
                "  END AS behavior_weight," +
                "  -- 时间衰减系数：根据行为时间与当前时间差计算" +
                "  CASE " +
                "    WHEN UNIX_TIMESTAMP() * 1000 - a.behavior_time <= " + Constant.TIME_WINDOW_30D + " THEN " + Constant.TIME_DECAY_30D +
                "    WHEN UNIX_TIMESTAMP() * 1000 - a.behavior_time <= " + Constant.TIME_WINDOW_60D + " THEN " + Constant.TIME_DECAY_60D +
                "    ELSE " + Constant.TIME_DECAY_OVER_60D +
                "  END AS time_decay," +
                "  a.ts " +
                "FROM ods_user_behavior a " +
                "WHERE " +
                "  a.user_id IS NOT NULL " + // 过滤用户ID为空的数据
                "  AND a.behavior_time > 0"; // 过滤行为时间异常的数据

        // 6. 执行DWD层数据处理任务
        tableEnv.executeSql(dwdEnrichmentSql);
        System.out.println("DWD层数据扩充任务启动成功");
        env.execute("UserBehaviorEnrichmentTask");
    }
}